package com.zlx.rulemk.marketing.dao;

import com.zlx.rulemk.marketing.beans.EventCombinationCondition;

import java.sql.Connection;
import java.sql.SQLException;
import com.zlx.rulemk.marketing.beans.BufferData;
import com.zlx.rulemk.marketing.beans.EventCombinationCondition;
import com.zlx.rulemk.marketing.buffer.BufferManager;
import com.zlx.rulemk.marketing.buffer.BufferManagerImpl;
import com.zlx.rulemk.marketing.constant.ConfigNames;
import com.zlx.rulemk.marketing.utils.EventIDEncodeUtil;
import com.zlx.rulemk.marketing.utils.EventUtil;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-08
 * @Description: clickhouse 数据查询 dao 类
 * @Version: 1.0
 */
@Slf4j
public class ClickHouseQuerier {

    Connection conn;
    BufferManager bufferManager;
    long bufferTtl;

    public ClickHouseQuerier(Connection conn){
        this.conn = conn;
        bufferManager = new BufferManagerImpl();
        Config config = ConfigFactory.load();
        bufferTtl = config.getLong(ConfigNames.REDIS_BUFFER_TTL);
    }

    /**
     * 返回的数据是  AADFURNFF,3
     *          事件code，匹配的次数
     * @param keyByValue
     * @param eventCombinationCondition
     * @return
     * @throws SQLException
     */
    public Tuple2<String,Integer> queryEventCombinationConditionCount
                                      (String keyByValue,
                                      EventCombinationCondition eventCombinationCondition
                                     ) throws SQLException {

        long queryRangeStart = eventCombinationCondition.getTimeRangeStart();
        long queryRangeEnd = eventCombinationCondition.getTimeRangeEnd();
        boolean needWholeStr = false;

        /**
         * 缓存在什么情况下有用？
         *   缓存数据的时间范围： [t3 -> t8]
         *   查询条件的时间范围：
         *[t3 -> t8]  直接用缓存的结果直接作为方法的返回值
         *[t3 -> t10] 判断缓存数据的count值是否 >= 条件的count阈值，如果成立，则直接返回缓存结果；否则，用  “缓存的结果+t8->t10”   作为整个返回结果
         *[t1 -> t8]  判断缓存数据的count值是否 >= 条件的count阈值，如果成立，则直接返回缓存结果；否则，用 “t1->t3+缓存的结果 ”  作为整个返回结果
         *[t1 -> t10] 判断缓存数据的count值是否 >= 条件的count阈值，如果成立，则直接返回缓存结果；否则，无用
         *
         *   如下逻辑实现，其实没有考虑一个问题：
         *      valueMap中，可能同时存在多个上述的区间范围可能性
         *      下面在遍历缓存数据的时候，可能遇到一个满足条件的就直接判断并往下走了
         *
         *   最好的实现应该是：
         *       比如，条件是  2->12
         *           而valuemap中可能存在的缓存数据：
         *                 2->10
         *                 4->12
         *                 4->10
         *                 1->13
         *      先遍历一遍valueMap，从中找到最优的 缓存区间数据
         *      然后再去判断，并往下走
         */

        String bufferKey = keyByValue + ":" + eventCombinationCondition.getCacheId();
        BufferData bufferData = bufferManager.getDataFromBuffer(bufferKey);
        Map<String, String> valueMap = bufferData.getValueMap();
        long current = System.currentTimeMillis();

        if(valueMap != null && valueMap.size() > 0 ){
            Set<String> keySet = valueMap.keySet();

            log.debug("下推CK之前，获取到redis缓存,redis大Key:{},redis中valueMap:{}",bufferKey,valueMap);
            /**
             * key
             * deviceId:cacheId
             *
             * valueMap
             * key      value
             * 2:6:20    AAABDCC
             * 2:8:10    ADDFSDG
             * 0:6:30    AAACCF
             */

            // 从redis缓存中 获取到了数据
            for (String key : keySet) {

                String[] split = key.split(":");
                long bufferStartTime = Long.parseLong(split[0]);
                long bufferEndTime = Long.parseLong(split[1]);
                long bufferInsertTime = Long.parseLong(split[2]);

                /**
                 * 判断缓存是否已过期，做清除动作
                 * TODO 这部分工作后期可以优化，不能影响业务处理性能，
                 */
                if(System.currentTimeMillis() - bufferInsertTime >= bufferTtl) {
                    bufferManager.delBufferEntry(bufferKey,key);
                    log.debug("dao-ck,清除过期缓存,bufferKey: {},key: {}",bufferKey,key);
                }

                // redis中缓存的数据串
                String bufferSeqStr = valueMap.get(key);
                // 匹配的次数
                int bufferCount = EventUtil.sequenceStrMatchRegexCount(bufferSeqStr, eventCombinationCondition.getMatchPattern());

                // 查询范围 和  缓存范围  完全相同，直接返回缓存中数据的结果
                // 缓存: |------|
                // 查询: |------|
                if(bufferStartTime == queryRangeStart && bufferEndTime == queryRangeEnd){
                    // 将原buffer数据从redis中清除，纯粹为了更新原缓存数据的insert时间戳
                    bufferManager.delBufferEntry(bufferKey,key);
                    HashMap<String, String> toPutMap = new HashMap<>();
                    String newSmartRediskey = bufferStartTime+":"+bufferEndTime+":"+current;
                    toPutMap.put(newSmartRediskey,bufferSeqStr);
                    bufferManager.putDataToBuffer(bufferKey,toPutMap);
                    log.debug("缓存时段=查询时段,reids大key:{},valueMap：{},更新redis小key的插入时间的值【{}-->{}】",bufferKey,valueMap,key,newSmartRediskey);
                    return Tuple2.of(bufferSeqStr,bufferCount);
                }

                // 左端点对其，且条件的时间范围包含缓存的时间范围
                // 缓存: |---origin---|
                // 查询: |---origin---|--right---|
                if(queryRangeStart == bufferStartTime && queryRangeEnd >= bufferEndTime){
                    int queryMinCount = eventCombinationCondition.getMinLimit();
                    // 是否满足阈值条件
                    // TODO 此处有可能返回一个非完整条件时间段的str,因此需要添加boolean参数
                    // TODO 这个地方有bug 还需要判断最大值是否满足 后期再做吧
                    if(bufferCount>=queryMinCount && !needWholeStr) {
                        return Tuple2.of(bufferSeqStr,bufferCount);
                    }else{
                        // 调整查询时间，去clickhouse中查一小段
                        String rightStr = getEventCombinationConditionStr(keyByValue, eventCombinationCondition ,bufferEndTime, queryRangeEnd);

                        // 将原buffer数据从redis中清除
                        bufferManager.delBufferEntry(bufferKey,key);
                        if(StringUtils.isNotBlank(rightStr)) {
                            HashMap<String, String> toPutMap = new HashMap<>();
                            // 写缓存，包含3种区间： 原buffer区间，  右边分段区间 ，  原buffer区间+右半边
                            // |---origin---|
                            //              |--right---|
                            // |---origin------right---|
                            toPutMap.put(bufferStartTime + ":" + bufferEndTime + ":" + current, bufferSeqStr);
                            toPutMap.put(bufferEndTime + ":" + queryRangeEnd + ":" + current, rightStr);
                            toPutMap.put(bufferStartTime + ":" + queryRangeEnd + ":" + current, bufferSeqStr + rightStr);
                            bufferManager.putDataToBuffer(bufferKey, toPutMap);

                            log.debug("缓存时段 右<=查询时段,原缓存bufferKey:{},key:{}, 原缓存value:{}, 写入value:{}", bufferKey, key, valueMap, toPutMap);
                        }
                        int totalCount = EventUtil.sequenceStrMatchRegexCount(bufferSeqStr + rightStr, eventCombinationCondition.getMatchPattern());
                        return Tuple2.of(bufferSeqStr+rightStr,totalCount);
                    }
                }

                // 右端点对其，且条件的时间范围包含缓存的时间范围
                // 缓存:             |------origin-----|
                // 查询: |---left----|------origin-----|
                if(queryRangeEnd == bufferEndTime && queryRangeStart < bufferStartTime){
                    int queryMinCount = eventCombinationCondition.getMinLimit();
                    // TODO 此处有可能返回一个非完整条件时间段的str
                    if(bufferCount>=queryMinCount && !needWholeStr) {
                        return Tuple2.of(bufferSeqStr,bufferCount);
                    }else{
                        // 调整查询时间，去clickhouse中查一小段
                        String leftStr = getEventCombinationConditionStr(keyByValue, eventCombinationCondition,bufferEndTime, queryRangeEnd);

                        // 将原buffer数据从redis中清除
                        bufferManager.delBufferEntry(bufferKey,key);

                        // 写缓存，包含3种区间： 原buffer区间，  左分段区间 ，  左分段+原buffer区间
                        //             |------origin-----|
                        // |---left----|
                        // |---left-----------origin-----|
                        if(StringUtils.isNotBlank(leftStr)) {
                            HashMap<String, String> toPutMap = new HashMap<>();
                            toPutMap.put(bufferStartTime + ":" + bufferEndTime + ":" + current, bufferSeqStr);
                            toPutMap.put(queryRangeStart + ":" + bufferStartTime + ":" + current, leftStr);
                            toPutMap.put(queryRangeStart + ":" + queryRangeEnd + ":" + current, leftStr + bufferSeqStr);
                            bufferManager.putDataToBuffer(bufferKey, toPutMap);
                            log.debug("查询时段 左< 缓存时段,原缓存bufferKey:{},key:{}, 原缓存value:{}, 写入value:{}", bufferKey, key, valueMap, toPutMap);
                        }
                        int totalCount = EventUtil.sequenceStrMatchRegexCount(leftStr + bufferSeqStr, eventCombinationCondition.getMatchPattern());
                        return Tuple2.of(leftStr+bufferSeqStr,totalCount);
                    }
                }

                // 缓存:       |-------|
                // 查询:  |----------------|
                if(queryRangeEnd >= bufferEndTime && queryRangeStart <= bufferStartTime){
                    int queryMinCount = eventCombinationCondition.getMinLimit();
                    if(bufferCount>=queryMinCount && !needWholeStr) {
                        // 将原buffer数据从redis中清除
                        bufferManager.delBufferEntry(bufferKey,key);
                        HashMap<String, String> toPutMap = new HashMap<>();
                        toPutMap.put(bufferStartTime+":"+bufferEndTime+":"+current,bufferSeqStr);
                        bufferManager.putDataToBuffer(bufferKey,toPutMap);
                        log.debug("查询时段 包含 缓存时段,原缓存bufferKey:{},key:{}, 原缓存value:{}, 写入value:{}",bufferKey,key,valueMap,toPutMap);
                        return Tuple2.of(bufferSeqStr,bufferCount);
                    }
                }

            }
        }else{
            log.debug("下推CK之前，【没有】获取到redis缓存,规则bufferKey:{}，全部数据去CK查询",bufferKey);
        }
        // 先查询到用户在组合条件中做过的事件的字符串序列 [AABAAAAAFFFCCCAAABBBCCCFFAAA]=> [11211111444333111222333441111]
        String eventSequenceStr = getEventCombinationConditionStr(keyByValue, eventCombinationCondition, queryRangeStart, queryRangeEnd);
        // 将查询结果写入缓存
        if(StringUtils.isNotBlank(eventSequenceStr)) {
            HashMap<String, String> toPutMap = new HashMap<>();
            toPutMap.put(queryRangeStart + ":" + queryRangeEnd + ":" + current, eventSequenceStr);
            bufferManager.putDataToBuffer(bufferKey, toPutMap);
            log.debug("缓存穿透,查询clickhouse,写入bufferKey:{},写入value:{}",bufferKey,toPutMap);
        }

        // 调用工具，来获取事件序列与正则表达式的匹配次数（既：组合条件发生的次数）
        int count = EventUtil.sequenceStrMatchRegexCount(eventSequenceStr, eventCombinationCondition.getMatchPattern());

        return  Tuple2.of(eventSequenceStr,count);

    }

    /**
     * 在clickhouse中，根据组合条件及查询的时间范围，得到返回结果的1213形式字符串序列
     * @param keyByValue
     * @param eventCombinationCondition
     * @return
     * @throws SQLException
     * 此方法改为private，确保没有任何上层模块可以调用本底层方法
     *
     * 这个代码里面的 queryRangeStart 是用户截取的开始时间，所以一定需要用户传入尽量的
     * queryRangeEnd 同理
     */
    private String getEventCombinationConditionStr(String keyByValue,
                                                   EventCombinationCondition eventCombinationCondition,
                                                   long queryRangeStart, long queryRangeEnd) throws SQLException {

        String querySql = eventCombinationCondition.getQuerySql();
        PreparedStatement preparedStatement = conn.prepareStatement(querySql);

        // 根据多个条件进行分组
        // TODO 多个条件分组的场景是什么？
        String[] keyParams = keyByValue.split(",");

        for(int i=0 ; i<keyParams.length;i++){
            preparedStatement.setString(i+1,keyParams[i]);
            log.debug("clickhouse查询前 ,preparestatement 设置的动态参数"+(i+1)+"：{}" ,keyParams[i]);
        }
        preparedStatement.setLong(keyParams.length+1,queryRangeStart);
        log.debug("clickhouse查询前 ,preparestatement 设置的开始时间参数：{}" ,queryRangeStart);
        preparedStatement.setLong(keyParams.length+2,queryRangeEnd);
        log.debug("clickhouse查询前 ,preparestatement 设置的结束时间参数：{}" ,queryRangeEnd);

        // 遍历ck返回的结果
        ResultSet resultSet = preparedStatement.executeQuery();
        StringBuilder sb = new StringBuilder();
        while(resultSet.next()){
            String eventId = resultSet.getString(1);
            // 根据eventId到组合条件的事件列表中找对应的索引号，来作为最终结果拼接
            // 换成业务的code
            String eventCode = EventIDEncodeUtil.getEventCode(eventId);
            sb.append(eventCode);
        }
        return sb.toString();

    }

}
